package com.haizhi.adapter;

import com.dinstone.beanstalkc.JobConsumer;
import com.haizhi.beanstalk.BeanstalkFactory;
import com.haizhi.enums.Province;
import com.haizhi.etl.JsonProcess;
import com.haizhi.exit.StopEnable;
import com.haizhi.factory.MongodbFactory;
import com.haizhi.factory.RedisFactory;
import com.haizhi.gsxt.GenRedisKey;
import com.haizhi.gsxt.ProvinceStore;
import com.haizhi.model.ClueEntity;
import com.haizhi.redis.RedisHandler;
import com.haizhi.util.ConfUtil;
import com.haizhi.util.JsonUtil;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.UpdateOneModel;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class ConsumerAdapter implements StopEnable {

    private static final Logger logger = LogManager.getLogger(ConsumerAdapter.class);

    //redis 缓存中过期时间
    private static final Integer CACHE_EXPIRE_SECONDS = 24 * 3600 * 15;

    //批量插入时间间隔
    private static final Integer BATCH_INSERT_TIME = 60000;

    //mongodb句柄
    private MongoDatabase mongoDatabase;

    //存储句柄管理
    private Map<String, ProvinceStore> storeMap = new HashMap<>();

    /**
     * redis 句柄
     */
    private RedisHandler redisHandler;

    /**
     * 发送的消息队列
     */
    private String beanstalkTube;
    private String beanstalkHost;
    private int beanstalkPort;

    /**
     * 运行标识
     */
    private boolean running = true;

    //初始化
    public ConsumerAdapter() throws Exception {


        //初始化MongoDB
        mongoDatabase = MongodbFactory.get("mongo.properties");

        //初始化redis
        redisHandler = RedisFactory.get("redis.properties");

        //初始化存储
        InitMongodbStore();

        //初始化消息队列
        InitBeanstalk();
    }

    private void InitMongodbStore() {
        //初始化存储
        for (Province province : Province.values()) {
            storeMap.put(province.getValue(), new ProvinceStore(province, mongoDatabase));
        }
    }

    private void InitBeanstalk() throws Exception {

        Properties property = ConfUtil.getProperties("beanstalk.properties");
        if (null == property) {
            throw new Exception("初始化beanstalk配置文件失败");
        }

        beanstalkTube = property.getProperty("beanstalk.tube");
        beanstalkHost = property.getProperty("beanstalk.host");
        beanstalkPort = Integer.valueOf(property.getProperty("beanstalk.port"));
        logger.info("消息队列初始化完成...");
    }

    //批量存储
    private void batchStoreTiming() {
        long startTime = System.currentTimeMillis();
        for (ProvinceStore provinceStore : storeMap.values()) {
            provinceStore.batchStoreTiming();
        }
        long endTime = System.currentTimeMillis();
        logger.info("定时存储耗时: {} ms", endTime - startTime);
    }

    /**
     * 消息监听
     */
    public void listen() {

        //获取消费者
        JobConsumer consumer = BeanstalkFactory.getConsumer(beanstalkHost, beanstalkPort, beanstalkTube);

        //获得redis句柄
        Jedis jedis = redisHandler.getJedisObject();

        long preTime = System.currentTimeMillis();

        logger.info("开始监听消息队列...");
        while (running) {
            try {


                String result = BeanstalkFactory.reserve(consumer, 3);

                //获取当前时间戳
                long currentTime = System.currentTimeMillis();

                //定期存储
                if (currentTime - preTime >= BATCH_INSERT_TIME) {
                    batchStoreTiming();
                    preTime = currentTime;
                }

                if (StringUtils.isEmpty(result)) {
                    continue;
                }

                ClueEntity clueEntity = JsonUtil.jsonToObject(result, ClueEntity.class);
                if (null == clueEntity) {
                    logger.error("转换map数据失败: json = {}", result);
                    continue;
                }

                // 获得当前数据省份信息
                String province = JsonProcess.parseProvince(clueEntity);

                //获取当前企业名称
                String company = JsonProcess.parseCompany(clueEntity);
                if (StringUtils.isEmpty(company)) {
                    logger.error("种子名称不符合规范: json = {}", result);
                    continue;
                }

                //判断当前企业是否已经在缓存中
                String scheduleKey = GenRedisKey.genScheduleKey(province, company);
                String redisValue = GenRedisKey.getRedisValue(province, company);
                if (redisHandler.isCacheExist(scheduleKey, jedis)) {
                    logger.warn("当前企业已在调度，15天内不再发起调度: province = {} company = {}",
                            province, company);
                    continue;
                }

                //存储调度记录到缓存
                redisHandler.saveCacheEx(scheduleKey, redisValue, CACHE_EXPIRE_SECONDS, jedis);

                logger.info("添加种子: province = {} company = {}", province, company);

                //打包数据格式
                UpdateOneModel<Document> updateDocument = JsonProcess.jsonToDocument(clueEntity, company);
                if (null == updateDocument) {
                    logger.error("提取 document 失败: json = {}", result);
                    continue;
                }


                //存储消费的数据
                storeMap.get(province).batchStoreDocument(updateDocument);
            } catch (Exception e) {
                logger.error("消息消费异常: ", e);
                //休眠一段时间
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    logger.error("休眠异常:", e1);
                }
            }
        }

        batchStoreTiming();
        logger.info("种子存储完成...");

        consumer.close();
        logger.info("消息队列关闭完成...");

        redisHandler.returnJedisOjbect(jedis);
        logger.info("redis关闭完成...");

        logger.info("安全退出...");
    }

    @Override
    public void shutdown() {
        running = false;
    }
}
